package com.message.xxljobhandler;

import com.base.model.R;
import com.message.feignclient.MediaClient;
import com.message.model.Message;
import com.message.service.MessageService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 小何
 * @Description: TODO
 * @DateTime: 2023/10/26 10:31
 **/
@Component
public class XxlJobHandler {
    private static Logger logger = LoggerFactory.getLogger(XxlJobHandler.class);

    @Autowired
    private MediaClient mediaClient;

    @Autowired
    private MessageService messageService;

    /**
     * 视频处理任务
     */
    @XxlJob("videoJobHandler")
    public void videoJobHandler() throws Exception {
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        //确定cpu核心数
        int processors = Runtime.getRuntime().availableProcessors();
        //查询待处理的任务
        List<Message> messages = messageService.selectTaskByShardIndex(shardIndex, shardTotal, processors);
        int size = messages.size();
        if (size == 0) {
            return;
        }
        //开启线程池
        ExecutorService executorService = Executors.newFixedThreadPool(size);
        //定义计数器
        CountDownLatch countDownLatch = new CountDownLatch(size);
        messages.forEach(message -> {
            executorService.execute(() -> {
                try {
                    String md5 = message.getMd5();
                    String suffix = message.getSuffix();
                    boolean transcode = messageService.transcode(md5, suffix);
                    if (transcode) {
                        messageService.updateTranscodeStatus(message.getId());
                        mediaClient.transcodeSuccess(message.getPublishId());
                    }
                } finally {
                    countDownLatch.countDown();
                }

            });
        });
        countDownLatch.await(30, TimeUnit.MINUTES);
    }
}
